package com.kg.fiber.service;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.util.LinkedList;
import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.commons.math3.transform.DftNormalization;
import org.apache.commons.math3.transform.FastFourierTransformer;
import org.apache.commons.math3.transform.TransformType;
import org.json.JSONObject;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import com.ivan.toolkit.common.Dispatcher;
import com.ivan.toolkit.common.Modules;
import com.ivan.toolkit.common.RunningInfo;
import com.kg.fiber.common.Count;
import com.kg.fiber.common.IntegerExpression;
import com.kg.fiber.common.Sum;
import com.kg.fiber.common.TimedDouble;
import com.kg.fiber.common.TimedDoubleBuffer;
import com.kg.fiber.domain.AppConfig;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.log4j.Log4j;

@Component
@Log4j
public class Boss extends Modules {
	
	private OriginSmooth originSmooth = new OriginSmooth();
	
	private OriginStore originStore = new OriginStore();
	
	private OriginSessionFactory orginFactory = new OriginSessionFactory();
	
	private OriginFFT originFFT = new OriginFFT();
	
	private OriginIntegration originIntegration = new OriginIntegration();
	
	private IntegrateStore integrateStore =  new IntegrateStore();
	
	private Judgment judgment = new Judgment();
	
	private StateMachine stateMachine = new StateMachine();
	
	private Reporter reporter = new Reporter();
	
	RunningInfo ri = new RunningInfo() {

		@Override
		protected void buildRunningInfo(ContentBuilder builder) {
			builder.buildSectionName("boss");
			builder.buildKeyValue("State", stateMachine.current.isWarning() ? "告警" : "正常");
			builder.buildKeyValue("Origin Count", originSmooth.count.get());
			builder.buildKeyValue("Integr Count", originIntegration.count.get());
			builder.buildKeyValue("Judge Count", judgment.values.size());
		}
		
	};
	
	@PostConstruct
	private void postConstruct() {
		this.register(originStore);
		this.register(integrateStore);
		this.initialize();
	}
	
	@PreDestroy
	private void preDestroy() {
		this.terminate();
	}
	
	@Scheduled(cron="0/2 * *  * * ? ")   
    public void doScheduled() {
		
		postWarning(stateMachine.current.isWarning());
		
	}
	
	public void postWarning(boolean isWarning) {
		
		AppConfig config = AppConfig.instance.get();
		
		String host = config.getReportHost();
		
		JSONObject json = new JSONObject();
		json.put("state", isWarning);
		
		reporter.post(host, json);
	}
	
	public void feed(int channel, TimedDoubleBuffer buffer) {
		if(channel == 0)
			originSmooth.feed(buffer);
	}

	class OriginSession {
		
		@Getter
		private List<TimedDouble> origin;
		
		@Getter
		private double integrated = Double.NaN;
		
		@Getter
		private AppConfig config;
		
		@Getter
		private long timestamp;
					
		private double[][] fft;
		

		public OriginSession(List<TimedDouble> origin, AppConfig config) {
			super();
			this.origin = origin;
			this.config = config;
			this.timestamp = origin.get(0).getTimestamp();
		}
		
		public void doFFT() {
			
			//StopWatch sw = new StopWatch();
			//sw.start();
			
			final double[][] dataRI = new double[2][origin.size()];
			
			for(int i = 0; i < origin.size(); i++) {
				dataRI[0][i++] = origin.get(i).getValue();
			}
			
			FastFourierTransformer.transformInPlace(dataRI, DftNormalization.STANDARD, TransformType.FORWARD);
			
			fft = dataRI;
			
			//sw.stop();
			//statisForFFT.addValue(sw.getTotalTimeMillis());
			
		}
		
		
		private double change(double r, double i) {
			return r * r + i * i;
		}
		
		public void doIntegration() {
			
			double result = 0.0;
			
			List<Integer> idx = IntegerExpression.parse(config.getIntegrateExpression());
			for(int i : idx) {
				if(i < fft[0].length) {
					result += change(fft[0][i], fft[1][i]);
				}
			}
			
			integrated = result;

		}

	}
	
	class OriginSmooth extends Dispatcher<TimedDoubleBuffer> {
		
		@Getter
		private long timeChaosCount = 0;
			
		private long last = 0L;
		
		//private long lastOutput = 0L;
			
		private List<TimedDoubleBuffer> unhandled = new LinkedList<TimedDoubleBuffer>();
		
		//@Getter
		//private Statistics statistics = new Statistics();
		
		private static final int splitPerMillis = 4;
		
		private Count count = new Count();

		@Override
		protected void dispatch(TimedDoubleBuffer event) {
			
			//if(resetor.reseting) return;
			
			long now = event.getTimestamp();
			
			if(last == 0L) {
				last = now;
				return;
			}
			
			unhandled.add(event);
			
			if(now < last) {
				log.warn("Time chaos, maybe pci clock card adjust.");
				log.warn("  last = " + last);
				log.warn("  now  = " + now);
				unhandled.clear();
				return;
			}
			
			long delta = now - last;
			
			// 控制收到多个包后再开始计算
			//if(delta < 1000) return;
				
			last = now;
			
			int len = 0;
			for(TimedDoubleBuffer b : unhandled) {
				len += b.getSize();
			}
			
			
			//List<TimedDouble> calculated = Lists.newLinkedList();
			
			long splitCount =  delta * splitPerMillis;
			
			// 计算
			{
				
				int part = len / (int)splitCount;
				
				int remainder = len % (int)splitCount;
				
				Sum sum = new Sum();
				
				long timestamp = now - delta + 1;
				
				long handleCount = 0;
									
				for(TimedDoubleBuffer b : unhandled) {
					
					for(int i = 0; i < b.getSize(); i++) {
						
						sum.add(b.get(i));					
						
						boolean feed = false;
						
						// 将余数平均计算到前面的点
						if(remainder > 0) {
							
							if(sum.getN() == (part + 1)) {
								feed = true;
								remainder--;
							}
							
						} else {
							
							if(sum.getN() == part) feed = true;
							
						}
					
						if(feed) {
							
							double value = sum.getMean();
							
							TimedDouble td = new TimedDouble(timestamp + (handleCount / splitPerMillis), value);
							
							originStore.feed(td);
							
							count.add();
														
							orginFactory.feed(td);
			
							sum.clear();
							
							handleCount++;
							
						}
						
					}
				
					TimedDoubleBuffer.pool.back(b);
				}
				
				unhandled.clear();
				
			}
			
		}

	}
	
	class OriginStore extends Store {

		@Override
		protected String getTag() {
			return "smooth";
		}
		
	}
	
	class OriginSessionFactory extends Dispatcher<TimedDouble> {
		
		private LinkedList<TimedDouble> tds = Lists.newLinkedList();

		@Override
		protected void dispatch(TimedDouble event) {
			
			tds.add(event);
			
			AppConfig config = AppConfig.instance.get();
			
			int size = config.getFFTSize();
			
			if(tds.size() >= size) {
				
				while(tds.size() > size) {
					tds.removeLast();
				}
				
				OriginSession os = new OriginSession(tds, config);
				
				originFFT.feed(os);
				
				tds = Lists.newLinkedList();
				
			}
					
		}

		
	}
	
	class OriginFFT extends Dispatcher<OriginSession> {
		

		@Override
		protected void dispatch(OriginSession event) {
			
			event.doFFT();
			
			originIntegration.feed(event);
					
		}

		
	}
	
	class OriginIntegration extends Dispatcher<OriginSession> {
		
		private Count count = new Count();
		

		@Override
		protected void dispatch(OriginSession event) {
			
			event.doIntegration();
			
			TimedDouble td = new TimedDouble(event.getTimestamp(), event.getIntegrated());
			
			integrateStore.feed(td);
			
			judgment.feed(td);
			
			count.add();
					
		}
		
	}
	
	class IntegrateStore extends Store {

		@Override
		protected String getTag() {
			return "integrate";
		}
		
	}
	
	@AllArgsConstructor
	class JudgeResult {
		
		@Getter
		private long timestamp;
		
		@Getter
		private boolean warning;
		
	}
	
	class Judgment extends Dispatcher<TimedDouble> {
		
		private List<TimedDouble> values = Lists.newLinkedList();
		
		@Override
		protected void dispatch(TimedDouble event) {
			
			values.add(event);
			
			AppConfig config = AppConfig.instance.get();
			
			int limit = config.getJudgeLimit();
			double threshold = config.getJudgeThreshold();
			int count = config.getJudgeCount();
			while(values.size() > count) {
				
				int upcount = 0;
				
				long timestamp = values.get(0).getTimestamp();
				
				for(int i = 0; i < count; i++) {
					
					if(values.get(i).getValue() > threshold)
						upcount++;
					
				}
				
				stateMachine.feed(new JudgeResult(timestamp, upcount > limit));
				
				values.remove(0);
				
			}
					
		}
		
	}
	
	class StateMachine extends Dispatcher<JudgeResult> {
		
		JudgeResult current = new JudgeResult(System.currentTimeMillis(), false); 
		
		@Override
		protected void dispatch(JudgeResult event) {
			
			current = event;
			
		}
		
	}
	
	class Reporter {
		
		public static final int ReportPort = 12345;
		
		DatagramSocket socket = null;
		
		public void post(String host, JSONObject value) {
			
			try {
				
				
				if(socket == null) {
					
					socket = new DatagramSocket();
					
				}
				
			
				String txt = value.toString();
				
				byte[] data = txt.getBytes(Charsets.UTF_8);
				
				DatagramPacket packet = new DatagramPacket(data, data.length);
				
				packet.setSocketAddress(new InetSocketAddress(host, ReportPort));
				
				this.socket.send(packet);
			
			} catch(Exception e) {
				
				log.error("Send json data failed.", e);
				
			}
			
		}
		
	}
	
}
